﻿using System;
using System.Linq;
using System.Collections.Generic;
using Gumini.JulaDB.Data;
using Gumini.JulaDB.Engine;
using Gumini.JulaDB.Expressions;
using Gumini.JulaDB.Optimizer;

namespace Gumini.JulaDB.Planner
{
    /// <summary>
    /// TODO: Update summary.
    /// </summary>
    internal class InnerJoinNode : JoinNode
    {
        private readonly InnerJoinClause _clause;

        public InnerJoinNode(PlanNode nodeA, PlanNode nodeB, List<ColumnPair> columnPairs)
            : this(new InnerJoinClause(
                left: nodeA,
                right: nodeB,
                columnPairs: columnPairs)) { }

        public InnerJoinNode(InnerJoinClause clause)
            : base(clause.LeftNode, clause.RightNode)
        {
            _clause = clause;
        }

        public override List<Row> Execute()
        {
            // this is a simple implementation of the hash-join algorithm

            var clause = _clause;
            bool reversed = false;
            // we reverse the join clause so left table is smaller and right is larger
            if (clause.LeftNode.Rows.Count > clause.RightNode.Rows.Count)
            {
                clause = clause.GetReversedClause();
                reversed = true;
            }

            var hashMapping = new RowMapping(true);
            foreach (var hashedRow in clause.LeftNode.Rows)
            {
                var context = hashedRow.GetEvaluationContext(clause.LeftNode.Heading);
                var keyValues = clause.ColumnPairs.
                    Select(p => p.Left.Evaluate(context)).ToList();
                hashMapping.Add(keyValues, hashedRow);
            }

            var result = new List<Row>();
            foreach (var keyRow in clause.RightNode.Rows)
            {
                var context = keyRow.GetEvaluationContext(clause.RightNode.Heading);
                var keyValues = clause.ColumnPairs.
                    Select(p => p.Right.Evaluate(context)).ToList();
                var key = new CompositeKey<TypedValue>(keyValues);
                if (hashMapping.ContainsKey(key))
                {
                    var matchedRows = hashMapping[key];
                    foreach (var matchedRow in matchedRows)
                    {
                        var newRow = reversed ?
                            new Row(keyRow, matchedRow) :
                            new Row(matchedRow, keyRow);
                        result.Add(newRow);
                    }
                }
            }
            
            return result;
        }

        protected override string GetNodeDescription()
        {
            return String.Format("JOIN (inner on {0})", _clause.ToString());
        }
    }
}
